package com.datareport.task;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.datareport.common.exception.BusinessException;
import com.datareport.compute.FlinkJobManager;
import com.datareport.domain.entity.StatRule;
import com.datareport.repository.StatRuleRepository;
import com.datareport.service.FlinkComputeService;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Flink计算任务调度器
 * 负责Flink作业的定时调度和监控
 */
@Slf4j
@Component
public class FlinkComputeScheduler {

    @Autowired
    private FlinkJobManager jobManager;

    @Autowired
    private FlinkComputeService computeService;

    @Autowired
    private StatRuleRepository ruleRepository;

    // 作业失败计数器
    private final Map<String, Integer> jobFailureCount = new ConcurrentHashMap<>();
    
    // 最大重试次数
    private static final int MAX_RETRY_TIMES = 3;
    
    // 作业重试冷却时间（分钟）
    private static final int RETRY_COOLING_MINUTES = 5;
    
    // 上次重试时间
    private final Map<String, LocalDateTime> lastRetryTime = new ConcurrentHashMap<>();

    /**
     * 初始化调度器
     */
    @PostConstruct
    public void init() {
        log.info("Flink计算任务调度器初始化完成");
        
        // 启动时自动恢复未完成的作业
        recoverRunningJobs();
    }

    /**
     * 定时检查作业状态
     * 每30秒执行一次
     */
    @Scheduled(fixedDelay = 30000)
    public void checkJobStatus() {
        try {
            log.debug("开始检查Flink作业状态");
            
            Map<String, FlinkJobManager.JobStatusInfo> jobStatuses = jobManager.getAllJobStatuses();
            
            for (Map.Entry<String, FlinkJobManager.JobStatusInfo> entry : jobStatuses.entrySet()) {
                String ruleId = entry.getKey();
                FlinkJobManager.JobStatusInfo statusInfo = entry.getValue();
                
                // 检查作业状态
                checkJobHealth(ruleId, statusInfo);
            }
            
            log.debug("Flink作业状态检查完成");
            
        } catch (Exception e) {
            log.error("检查Flink作业状态失败", e);
        }
    }

    /**
     * 定时清理过期结果数据
     * 每天凌晨2点执行
     */
    @Scheduled(cron = "0 0 2 * * ?")
    public void cleanupExpiredResults() {
        try {
            log.info("开始清理过期的规则计算结果数据");
            
            // 获取所有启用的规则
            List<StatRule> rules = ruleRepository.selectList(
                    new QueryWrapper<StatRule>().eq("status", 1)
            );

            // 清理每个规则的过期数据（默认保留30天）
            for (StatRule rule : rules) {
                try {
                    computeService.cleanupRuleResults(rule.getRuleId(), 30);
                } catch (Exception e) {
                    log.error("清理规则结果数据失败: ruleId={}", rule.getRuleId(), e);
                }
            }
            
            log.info("过期规则计算结果数据清理完成");
            
        } catch (Exception e) {
            log.error("清理过期规则计算结果数据失败", e);
        }
    }

    /**
     * 定时验证规则配置
     * 每天凌晨3点执行
     */
    @Scheduled(cron = "0 0 3 * * ?")
    public void validateRuleConfigs() {
        try {
            log.info("开始验证规则配置");
            
            // 获取所有规则
            List<StatRule> rules = ruleRepository.selectList(null);
            
            for (StatRule rule : rules) {
                try {
                    boolean isValid = computeService.validateRuleConfig(rule);
                    if (!isValid) {
                        log.warn("规则配置验证失败: ruleId={}, name={}", rule.getRuleId(), rule.getName());
                        
                        // 如果规则正在运行，停止作业
                        if (jobManager.isJobRunning(rule.getRuleId())) {
                            jobManager.stopJob(rule.getRuleId());
                            log.info("停止配置验证失败的作业: ruleId={}", rule.getRuleId());
                        }
                    }
                } catch (Exception e) {
                    log.error("验证规则配置失败: ruleId={}", rule.getRuleId(), e);
                }
            }
            
            log.info("规则配置验证完成");
            
        } catch (Exception e) {
            log.error("验证规则配置失败", e);
        }
    }

    /**
     * 定时重启失败的作业
     * 每5分钟执行一次
     */
    @Scheduled(fixedDelay = 300000)
    public void restartFailedJobs() {
        try {
            log.debug("开始重启失败的Flink作业");
            
            Map<String, FlinkJobManager.JobStatusInfo> jobStatuses = jobManager.getAllJobStatuses();
            
            for (Map.Entry<String, FlinkJobManager.JobStatusInfo> entry : jobStatuses.entrySet()) {
                String ruleId = entry.getKey();
                FlinkJobManager.JobStatusInfo statusInfo = entry.getValue();
                
                // 检查是否需要重启
                if (shouldRestartJob(ruleId, statusInfo)) {
                    try {
                        restartJobWithRetry(ruleId);
                    } catch (Exception e) {
                        log.error("重启失败作业失败: ruleId={}", ruleId, e);
                    }
                }
            }
            
            log.debug("失败的Flink作业重启完成");
            
        } catch (Exception e) {
            log.error("重启失败的Flink作业失败", e);
        }
    }

    /**
     * 检查作业健康状态
     * @param ruleId 规则ID
     * @param statusInfo 作业状态信息
     */
    private void checkJobHealth(String ruleId, FlinkJobManager.JobStatusInfo statusInfo) {
        try {
            switch (statusInfo.getStatus()) {
                case FAILED:
                    log.error("作业执行失败: ruleId={}, message={}", ruleId, statusInfo.getMessage());
                    handleJobFailure(ruleId, statusInfo.getMessage());
                    break;
                    
                case CANCELED:
                    log.warn("作业被取消: ruleId={}, message={}", ruleId, statusInfo.getMessage());
                    // 如果是意外取消，可以尝试重启
                    if (!"作业已停止".equals(statusInfo.getMessage())) {
                        handleJobFailure(ruleId, "作业意外取消");
                    }
                    break;
                    
                case FINISHED:
                    log.info("作业正常完成: ruleId={}", ruleId);
                    // 清理失败计数
                    jobFailureCount.remove(ruleId);
                    lastRetryTime.remove(ruleId);
                    break;
                    
                case RUNNING:
                    // 作业正常运行，无需处理
                    break;
                    
                default:
                    log.debug("作业状态: ruleId={}, status={}", ruleId, statusInfo.getStatus());
                    break;
            }
            
        } catch (Exception e) {
            log.error("检查作业健康状态失败: ruleId={}", ruleId, e);
        }
    }

    /**
     * 处理作业失败
     * @param ruleId 规则ID
     * @param errorMessage 错误消息
     */
    private void handleJobFailure(String ruleId, String errorMessage) {
        try {
            // 增加失败计数
            int failures = jobFailureCount.getOrDefault(ruleId, 0) + 1;
            jobFailureCount.put(ruleId, failures);
            
            log.warn("作业失败计数增加: ruleId={}, failures={}", ruleId, failures);
            
            // 如果失败次数过多，发送告警
            if (failures >= MAX_RETRY_TIMES) {
                log.error("作业连续失败次数过多，停止自动重试: ruleId={}, failures={}", ruleId, failures);
                // 这里可以添加告警通知逻辑
                sendAlarm(ruleId, "作业连续失败次数过多，停止自动重试");
            }
            
        } catch (Exception e) {
            log.error("处理作业失败失败: ruleId={}", ruleId, e);
        }
    }

    /**
     * 判断是否应重启作业
     * @param ruleId 规则ID
     * @param statusInfo 作业状态信息
     * @return 是否应重启
     */
    private boolean shouldRestartJob(String ruleId, FlinkJobManager.JobStatusInfo statusInfo) {
        // 检查作业状态是否为失败或取消
        if (statusInfo.getStatus() != org.apache.flink.api.common.JobStatus.FAILED && 
            statusInfo.getStatus() != org.apache.flink.api.common.JobStatus.CANCELED) {
            return false;
        }
        
        // 检查失败次数
        int failures = jobFailureCount.getOrDefault(ruleId, 0);
        if (failures >= MAX_RETRY_TIMES) {
            return false;
        }
        
        // 检查重试冷却时间
        LocalDateTime lastRetry = lastRetryTime.get(ruleId);
        if (lastRetry != null) {
            LocalDateTime now = LocalDateTime.now();
            if (lastRetry.plusMinutes(RETRY_COOLING_MINUTES).isAfter(now)) {
                return false;
            }
        }
        
        return true;
    }

    /**
     * 重启作业（带重试逻辑）
     * @param ruleId 规则ID
     */
    private void restartJobWithRetry(String ruleId) {
        try {
            log.info("开始重启失败的作业: ruleId={}", ruleId);
            
            // 记录重试时间
            lastRetryTime.put(ruleId, LocalDateTime.now());
            
            // 验证规则配置
            StatRule rule = ruleRepository.selectById(ruleId);
            boolean isValid = computeService.validateRuleConfig(rule);
            if (!isValid) {
                log.warn("作业配置验证失败，跳过重启: ruleId={}", ruleId);
                return;
            }
            
            // 重启作业
            jobManager.restartJob(ruleId);
            
            log.info("作业重启成功: ruleId={}", ruleId);
            
        } catch (Exception e) {
            log.error("重启作业失败: ruleId={}", ruleId, e);
            throw new BusinessException("重启作业失败: " + e.getMessage());
        }
    }

    /**
     * 恢复运行中的作业
     * 在系统启动时自动恢复之前未正常停止的作业
     */
    private void recoverRunningJobs() {
        try {
            log.info("开始恢复运行中的Flink作业");
            
            // 获取所有启用的规则
            List<StatRule> rules = ruleRepository.selectList(
                    new QueryWrapper<StatRule>().eq("status", 1)
            );
            
            for (StatRule rule : rules) {
                try {
                    String ruleId = rule.getRuleId();
                    FlinkJobManager.JobStatusInfo statusInfo = jobManager.getJobStatus(ruleId);
                    
                    // 如果作业状态为运行中，尝试恢复
                    if (statusInfo.isRunning()) {
                        log.info("发现运行中的作业，尝试恢复: ruleId={}", ruleId);
                        
                        // 验证配置
                        boolean isValid = computeService.validateRuleConfig(rule);
                        if (!isValid) {
                            log.warn("作业配置验证失败，停止恢复: ruleId={}", ruleId);
                            jobManager.stopJob(ruleId);
                            continue;
                        }
                        
                        // 如果作业实际未运行，重启作业
                        if (!isJobActuallyRunning(statusInfo)) {
                            log.info("作业实际未运行，重新启动: ruleId={}", ruleId);
                            jobManager.restartJob(ruleId);
                        }
                    }
                    
                } catch (Exception e) {
                    log.error("恢复作业失败: ruleId={}", rule.getRuleId(), e);
                }
            }
            
            log.info("Flink作业恢复完成");
            
        } catch (Exception e) {
            log.error("恢复运行中的Flink作业失败", e);
        }
    }

    /**
     * 检查作业是否实际在运行
     * @param statusInfo 作业状态信息
     * @return 是否实际在运行
     */
    private boolean isJobActuallyRunning(FlinkJobManager.JobStatusInfo statusInfo) {
        try {
            // 这里可以添加更详细的检查逻辑
            // 例如通过Flink REST API检查作业的实际状态
            return statusInfo.isRunning();
        } catch (Exception e) {
            log.error("检查作业实际运行状态失败: ruleId={}", statusInfo.getRuleId(), e);
            return false;
        }
    }

    /**
     * 发送告警通知
     * @param ruleId 规则ID
     * @param message 告警消息
     */
    private void sendAlarm(String ruleId, String message) {
        try {
            log.error("发送告警通知: ruleId={}, message={}", ruleId, message);
            
            // 这里可以集成实际的告警通知系统
            // 例如：钉钉、企业微信、邮件等
            
            // 示例：记录告警日志
            log.warn("ALARM - 规则计算作业异常: ruleId={}, message={}", ruleId, message);
            
        } catch (Exception e) {
            log.error("发送告警通知失败: ruleId={}", ruleId, e);
        }
    }

    /**
     * 获取调度器状态
     * @return 调度器状态信息
     */
    public Map<String, Object> getSchedulerStatus() {
        try {
            Map<String, Object> status = new ConcurrentHashMap<>();
            
            // 作业状态统计
            Map<String, FlinkJobManager.JobStatusInfo> jobStatuses = jobManager.getAllJobStatuses();
            long runningCount = jobStatuses.values().stream()
                .filter(FlinkJobManager.JobStatusInfo::isRunning)
                .count();
            long failedCount = jobStatuses.values().stream()
                .filter(info -> info.getStatus() == org.apache.flink.api.common.JobStatus.FAILED)
                .count();
            
            status.put("totalJobs", jobStatuses.size());
            status.put("runningJobs", runningCount);
            status.put("failedJobs", failedCount);
            status.put("failureCount", jobFailureCount);
            status.put("lastRetryTime", lastRetryTime);
            status.put("schedulerActive", true);
            
            return status;
            
        } catch (Exception e) {
            log.error("获取调度器状态失败", e);
            Map<String, Object> status = new ConcurrentHashMap<>();
            status.put("schedulerActive", false);
            status.put("error", e.getMessage());
            return status;
        }
    }
}